package collector.metaexec;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import collector.core.util.DateUtil;
import collector.httpd.metad.report.ReportRequest;
import collector.task.JobTaskVO;

public class MetaExecMon implements MetaExec {
	private Logger logger = Logger.getLogger(getClass());
	private Map<String, Long> hostMap = new ConcurrentHashMap<String, Long>();
	private ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(2);
	private int checkDelay = 5;	// sec
	private int checkTime = 5;	// sec
	private Runnable hostChecker = new Runnable() {
		public void run() {
			checkDeadHost();
		}
	};
	private void checkDeadHost() {
		if(hostMap==null) return;
		long currTime = DateUtil.getTimelong();
		Iterator<String> it = hostMap.keySet().iterator();
		while(it.hasNext()) {
			String key = (String) it.next();
			long lastTime = (Long) hostMap.get(key);
			if(lastTime + checkTime < currTime) {
				hostMap.remove(key);
				logger.info(hostMap.size() + ", host is down: " + key);
			}
		}
	}
	@SuppressWarnings("unused")
	private void printHost() {
		if(true) return; // FIXME::Dead Code for printHost()
		StringBuilder strb = new StringBuilder();
		strb.append("\n---------- currunt host ----------");
		if(hostMap==null) {
			strb.append("\n0, host is running");
			logger.info(strb.toString());
			return;
		}
		strb.append("\ntotal: " + hostMap.size());
		long currTime = DateUtil.getTimelong();
		Iterator<String> it = hostMap.keySet().iterator();
		while(it.hasNext()) {
			String key = (String) it.next();
			long lastTime = (Long) hostMap.get(key);
			strb.append("\n" + key
					+ ", " + (currTime - lastTime));
		}
		logger.info(strb.toString());
	}
	public class HostInfo {
		@SuppressWarnings("unused")
		private String hostName;
		@SuppressWarnings("unused")
		private long lastTime;
	}

	@SuppressWarnings("unused")
	private boolean isHostAlive() {
		int ttl = 60;
		int hostTn = 0, hostTmax = 0;
		int clusterLocalTime = 0, hostReported = 0;
		if(hostTn>0 && hostTmax>0)
			return hostTn <= hostTmax*4;
		// old
		return Math.abs(clusterLocalTime - hostReported) <= (ttl*4);
	}
	
	public MetaExecMon() {
		exec.scheduleWithFixedDelay(hostChecker, 0, checkDelay, TimeUnit.SECONDS);
		exec.scheduleAtFixedRate(new Runnable() {
			public void run() {
				printHost();
			}
		}, 0, 10, TimeUnit.SECONDS);
	}

	public void initialize() {
	}

	public void destroy() {
	}

	public void execMeta(ReportRequest reqvo, Object vo) {
		if(!(vo instanceof JobTaskVO)) return;
		
		handleHeartbeatData(reqvo, (JobTaskVO) vo);
	}

	private void handleHeartbeatData(ReportRequest reqvo, JobTaskVO vo) {
		String key = reqvo.getMonCluster() + "/" + reqvo.getServerNodeId();
		if(!hostMap.containsKey(key)) {
			logger.info((hostMap.size()+1) + ", host is up: " + key + ", " + reqvo.getServerHostName());
		}
		hostMap.put(key, reqvo.getServerTime());
	}


}
